package com.atguigu.gmall.realtime.dws.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.base.BaseApp;
import com.atguigu.gmall.realtime.common.bean.TrafficPageViewBean;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.function.BeanToJsonStrMapFunction;
import com.atguigu.gmall.realtime.common.util.DateFormatUtil;
import com.atguigu.gmall.realtime.common.util.FlinkSinkUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * @author Felix
 * @date 2024/10/12
 * 版本、渠道、地区、新老访客聚合统计
 * 需要启动的进程
 *      zk、kafka、flume、doris、DwdBaseLog、DwsTrafficVcChArIsNewPageViewWindow
 */
public class DwsTrafficVcChArIsNewPageViewWindow extends BaseApp {

    public static void main(String[] args) {
        new DwsTrafficVcChArIsNewPageViewWindow().start(
                10022,
                4,
                "dws_traffic_vc_ch_ar_is_new_page_view_window",
                Constant.TOPIC_DWD_TRAFFIC_PAGE
        );

    }
    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> kafkaStrDS) {
        //TODO 1.对流中数据进行类型转换    jsonStr->jsonObj
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.map(JSON::parseObject);
        //jsonObjDS.print();
        //TODO 2.按照mid进行分组
        KeyedStream<JSONObject, String> midKeyedDS = jsonObjDS.keyBy(jsonObj -> jsonObj.getJSONObject("common").getString("mid"));
        //TODO 3.再次对流中数据进行类型转换  jsonObj->实体类对象
        SingleOutputStreamOperator<TrafficPageViewBean> beanDS = midKeyedDS.process(
                new KeyedProcessFunction<String, JSONObject, TrafficPageViewBean>() {
                    //声明状态
                    ValueState<String> lastVisitDateState;
                    //状态的初始化
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<String> valueStateDescriptor
                                = new ValueStateDescriptor<String>("lastVisitDateState",String.class);
                        valueStateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1)).build());
                        lastVisitDateState = getRuntimeContext().getState(valueStateDescriptor);
                    }

                    @Override
                    public void processElement(JSONObject jsonObj, KeyedProcessFunction<String, JSONObject, TrafficPageViewBean>.Context ctx, Collector<TrafficPageViewBean> out) throws Exception {
                        JSONObject commonJsonObj = jsonObj.getJSONObject("common");
                        String vc = commonJsonObj.getString("vc");
                        String ch = commonJsonObj.getString("ch");
                        String ar = commonJsonObj.getString("ar");
                        String isNew = commonJsonObj.getString("is_new");

                        //~~~独立访客数~~~
                        Long uvCt = 0L;
                        //获取当前设备上次访问日期
                        String lastVisitDate = lastVisitDateState.value();
                        //获取当前访问日期
                        Long ts = jsonObj.getLong("ts");
                        String curVisitDate = DateFormatUtil.tsToDate(ts);
                        if(StringUtils.isEmpty(lastVisitDate) || !lastVisitDate.equals(curVisitDate)){
                            uvCt = 1L;
                            lastVisitDateState.update(curVisitDate);
                        }
                        //~~~会话数~~~
                        JSONObject pageJsonObj = jsonObj.getJSONObject("page");
                        String lastPageId = pageJsonObj.getString("last_page_id");
                        Long svCt = StringUtils.isEmpty(lastPageId) ? 1L : 0L;


                        TrafficPageViewBean viewBean = new TrafficPageViewBean(
                                "",
                                "",
                                "",
                                vc,
                                ch,
                                ar,
                                isNew,
                                uvCt,
                                svCt,
                                1L,
                                pageJsonObj.getLong("during_time"),
                                ts
                        );

                        out.collect(viewBean);
                    }
                }
        );
        //beanDS.print();

        //TODO 4.指定Watermark的生成策略以及提取事件时间字段
        /*
        前提：事件时间语义
            水位线Watermark是一个逻辑时钟，是一个衡量事件时间进展的标记
            也会随着流的流动向下游传递
            主要用于触发事件时间窗口计算、关闭以及事件时间定时器执行
            水位线生成是递增永不减小
            在FlinkAPI内部，提供了2种水位线生成策略
            单调递增、有界乱序
            单调递增是有界乱序的子类，是有界乱序的特殊情况(乱序程度是0)
            底层Watermark的生成过程
            类  implements WatermarkGenerator{
                流中每来一条数据：比较事件时间最大值
                周期性执行的方法(200ms):        计算水位线--最大事件时间-乱序程度-1ms
            }
        水位线的传递
            上游是1个并行度，下游是n个并行度   广播
            上游是n个并行度，下游是1个并行度   将上游各个并行度水位值取最小
            上游是n个并行度，下游是n个并行度   先广播再取最小
        */
        SingleOutputStreamOperator<TrafficPageViewBean> withWatermarkDS = beanDS.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<TrafficPageViewBean>forMonotonousTimestamps()
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<TrafficPageViewBean>() {
                                    @Override
                                    public long extractTimestamp(TrafficPageViewBean viewBean, long recordTimestamp) {
                                        return viewBean.getTs();
                                    }
                                }
                        )
        );

        //TODO 5.按照统计的维度进行分组
        /*
        在开窗前是否进行keyBy
        keyBy-针对keyBy后的每一个组，独立开窗，窗口间相互不影响
        noKeyBy-针对整条流进行开窗，并行度变为1
        */
        KeyedStream<TrafficPageViewBean, Tuple4<String, String, String, String>> dimKeyedDS = withWatermarkDS.keyBy(
                new KeySelector<TrafficPageViewBean, Tuple4<String, String, String, String>>() {
                    @Override
                    public Tuple4<String, String, String, String> getKey(TrafficPageViewBean viewBean) throws Exception {
                        return Tuple4.of(
                                viewBean.getVc(),
                                viewBean.getCh(),
                                viewBean.getAr(),
                                viewBean.getIsNew()
                        );
                    }
                }
        );
        //TODO 6.开窗(滚动事件时间窗口)
        /*
        窗口对象什么时候创建
            当属于这个窗口的第一个元素到来的时候，创建窗口对象
        窗口的起始时间向下取整   结束时间=起始时间 + 窗口大小    最大时间=结束时间 - 1ms
        窗口什么时候触发计算
            水位线>= 窗口最大时间
        窗口什么时候关闭
            水位线 >= 窗口最大事件 + 允许迟到时间
        在Flink中，对于迟到数据如何处理
            在指定水位线的时候，设置乱序程序
            在开窗的时候，设置窗口的允许迟到时间
            侧输出流
        */
        WindowedStream<TrafficPageViewBean, Tuple4<String, String, String, String>, TimeWindow> windowDS
                = dimKeyedDS.window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)));

        //TODO 7.聚合
        /*
        *  对窗口的数据进行处理
        *       增量处理
        *           好处:不会缓存窗口数据，省空间     缺点：获取不到窗口更全面的信息
        *           reduce
        *               如果窗口中元素类型、累加类型、向下游传递的类型一致
        *           aggregate
        *               如果窗口中元素类型、累加类型、向下游传递的类型不一致
        *       全量处理
        *           好处:获取窗口更全面的信息        缺点：会缓存窗口数据，费空间
        *           apply
        *               方法参数接收的是窗口对象
        *           process
        *               方法参数接收的是上下位对象，上下文对象除了可以获取窗口对象之外，还可以获取其它的一些信息，更底层
        * */
        SingleOutputStreamOperator<TrafficPageViewBean> reduceDS = windowDS.reduce(
                new ReduceFunction<TrafficPageViewBean>() {
                    @Override
                    public TrafficPageViewBean reduce(TrafficPageViewBean value1, TrafficPageViewBean value2) throws Exception {
                        value1.setPvCt(value1.getPvCt() + value2.getPvCt());
                        value1.setUvCt(value1.getUvCt() + value2.getUvCt());
                        value1.setDurSum(value1.getDurSum() + value2.getDurSum());
                        value1.setSvCt(value1.getSvCt() + value2.getSvCt());
                        return value1;
                    }
                },
                new WindowFunction<TrafficPageViewBean, TrafficPageViewBean, Tuple4<String, String, String, String>, TimeWindow>() {
                    @Override
                    public void apply(Tuple4<String, String, String, String> stringStringStringStringTuple4, TimeWindow window, Iterable<TrafficPageViewBean> input, Collector<TrafficPageViewBean> out) throws Exception {
                        TrafficPageViewBean viewBean = input.iterator().next();

                        viewBean.setStt(DateFormatUtil.tsToDateTime(window.getStart()));
                        viewBean.setEdt(DateFormatUtil.tsToDateTime(window.getEnd()));
                        viewBean.setCur_date(DateFormatUtil.tsToDate(window.getStart()));
                        //将对象向下游传递
                        out.collect(viewBean);
                    }
                }
        );

        //TODO 8.将聚合的结果写到Doris表中
        reduceDS.print();
        reduceDS
                //在向Doris写入数据前，需要将流中对象转换为json字符串
                .map(new BeanToJsonStrMapFunction<>())
                .sinkTo(FlinkSinkUtil.getDorisSink("dws_traffic_vc_ch_ar_is_new_page_view_window"));

    }
}
